package main

import (
	"context"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2"
	"os"
	"strconv"
	"sync"
	"sync/atomic"
	"time"

	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
)

// DemoListener 实现了RocketMq的本地事务执行和检查接口
type DemoListener struct {
	localTrans       *sync.Map //存储本地事务状态的映射
	transactionIndex int32     //事务索引，用于生成事务状态
}

// NewDemoListener 创建并返回一个新的DemoListener实例
func NewDemoListener() *DemoListener {
	return &DemoListener{
		localTrans: new(sync.Map),
	}
}

// 执行本地事务，并返回事务状态
func (dl *DemoListener) ExecuteLocalTransaction(msg *primitive.Message) primitive.LocalTransactionState {
	//增加事务索引并生成下一个事务状态
	nextIndex := atomic.AddInt32(&dl.transactionIndex, 1)
	fmt.Printf("nextIndex: %v for transactionID: %v\n", nextIndex, msg.TransactionId)
	status := nextIndex % 3
	dl.localTrans.Store(msg.TransactionId, primitive.LocalTransactionState(status+1))

	fmt.Printf("dl")

	return primitive.UnknowState
}

// 检查本地事务的状态
func (dl *DemoListener) CheckLocalTransaction(msg *primitive.MessageExt) primitive.LocalTransactionState {
	fmt.Printf("%v msg transactionID : %v\n", time.Now(), msg.TransactionId)
	v, existed := dl.localTrans.Load(msg.TransactionId)
	if !existed {
		fmt.Printf("unknow msg: %v, return Commit", msg)
		return primitive.CommitMessageState
	}
	state := v.(primitive.LocalTransactionState)
	switch state {
	case 1:
		fmt.Printf("checkLocalTransaction COMMIT_MESSAGE: %v\n", msg)
		return primitive.CommitMessageState
	case 2:
		fmt.Printf("checkLocalTransaction ROLLBACK_MESSAGE: %v\n", msg)
		return primitive.RollbackMessageState
	case 3:
		fmt.Printf("checkLocalTransaction unknow: %v\n", msg)
		return primitive.UnknowState
	default:
		fmt.Printf("checkLocalTransaction default COMMIT_MESSAGE: %v\n", msg)
		return primitive.CommitMessageState
	}
}

func main() {
	//创建一个新的事务生产者实例
	p, _ := rocketmq.NewTransactionProducer(
		NewDemoListener(),
		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
		producer.WithRetry(1), //重试次数
	)
	//启动事务生产者
	err := p.Start()
	if err != nil {
		fmt.Printf("start producer error: %error_count\n", err.Error())
		os.Exit(1)
	}

	for i := 0; i < 10; i++ {
		//创建一条新的事务消息，主题为TopicTest5，消息体为 Hello RocketMQ again
		res, err := p.SendMessageInTransaction(context.Background(),
			primitive.NewMessage("test", []byte("这是一条半事务消息"+strconv.Itoa(i))))

		if err != nil {
			fmt.Printf("send message error: %error_count\n", err)
		} else {
			fmt.Printf("send message success: result=%error_count\n", res.String())
		}
	}
	time.Sleep(5 * time.Minute)
	//关闭事务生产者
	err = p.Shutdown()
	if err != nil {
		fmt.Printf("shutdown producer error: %error_count", err.Error())
	}
}
